/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.worker;

import cn.ac.iie.di.ban.ycl.idgen.IDSHAGeneratorImpl;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class ReceiveHandler extends REAbstractReceiveMessageHandler<REMessage> {

    static private Logger logger = Logger.getLogger(ReceiveHandler.class.getName());

    static private final HashMap<String, REFieldType> types;
    BlockingQueue<HashMap<String, Object>> queue;

    static {
        types = new HashMap<>();
        types.put("_id", REFieldType.String);
        types.put("_key", REFieldType.String);
        types.put("_spec", REFieldType.String);
        types.put("_ch", REFieldType.Long);
        types.put("wx_nap", REFieldType.Long);
    }
    static private final String[] lookupFields = {"_id", "_key", "_spec", "_ch", "wx_sfrom", "wx_nap"};

    private boolean stopped;
    private final AtomicInteger aliveCount;
    private final Random rand;

    public ReceiveHandler(BlockingQueue<HashMap<String, Object>> queue) {
        this.rand = new Random(System.currentTimeMillis());
        stopped = false;
        aliveCount = new AtomicInteger(0);
        this.queue = queue;
    }

    static private IDSHAGeneratorImpl idgenerator = new IDSHAGeneratorImpl();

    static private String getId(String wx_uid) {
        try {
            return idgenerator.getId("40", "U-WX03-A", "Wechat", "Wechat", wx_uid);
        } catch (Exception ex) {
            return null;
        }
    }

    @Override
    public boolean handle(REMessage message) {
//        logger.info("start consum one message");
        if (false) {
            return true;
        }
        //env
        if (stopped) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException ex) {
            }
            return false;
        }
        aliveCount.getAndIncrement();
        logger.info("receive record size: " + message.data.size());
        //handle
        try {
            String[] keys = message.colNames;
//            logger.info("receive message size = " + message.data.size());
            for (Object[] rec : message.data) {
                HashMap<String, Object> line = new HashMap<>(rec.length + 5);
                int idx = -1;
                for (String key : keys) {
                    line.put(key, rec[++idx]);
                }
                while (true) {
                    try {
                        queue.put(line);
//                        logger.info("_id:"+line.get("_id"));
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("put into queue error", ex);
                    }
                }
            }

        } catch (Exception ex) {
            logger.error("some thing error occur when do delayed init!", ex);
        } finally {
            aliveCount.getAndDecrement();
        }
        return true;
    }

    public void stopGracefully() {
        stopped = true;
        while (aliveCount.get() > 0) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException ex) {
                logger.warn("sleep interrupted.", ex);
            }
        }
    }

}
